package BDMSource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class BDM {
	
	 public static IntWritable partitionIndex;
	 public static long sumPairs;
	 public static ArrayList<BDMData> BDMData = new ArrayList<BDMData>();
	 public static int blockingPos = 0;
	 public static HashMap<String, Integer> blockingkeyIndexer = new HashMap<String, Integer>(); 
	
	 public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
		 
		 private final static IntWritable one = new IntWritable(1);
		 private Text word = new Text();
	        
		 @Override
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			 word.set(value);
			 String blockingKey = computeKey(word.toString().toUpperCase());
			 word.set(blockingKey + "." + getMapIndex(context.getInputSplit().toString()));
			 context.write(word, one);       
		 }
		 
		 private int getMapIndex(String taskAttemptID){
			 taskAttemptID = taskAttemptID.substring(0, taskAttemptID.lastIndexOf("."));
			 return Integer.parseInt(taskAttemptID.substring(taskAttemptID.lastIndexOf("_")+1, taskAttemptID.length()));	
		 }
		 
		 public String computeKey(String entity){
			 if (entity.length() < 3) return ">>>";
			 return entity.substring(0, 3).replace(".", ",");	
		 }
	    
	 } 
		        
	 public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
	
		 @Override
		public void reduce(Text key, Iterable<IntWritable> values, Context context) 
	      	throws IOException, InterruptedException {
			 int sum = 0;
			 for (IntWritable val : values) {
				 sum += val.get();
			 }
			 context.write(key, new IntWritable(sum));
		 }
	 }
	 
	 public static class Map_configure extends Configuration{
			
	    public void map_configure(Map m, Reduce r, IntWritable partIndex) throws IOException, InterruptedException {
	    	//Store
	    	partitionIndex = partIndex;
	    }  
	 }
	 
	 public static long pairs(){
    	return BDM.sumPairs;
	 }
	 
	 public static ArrayList<BDMData> readBDM(){
		 return BDMData;
	 }
	 
	 public static int numBlocks(){
		 return BDMData.size();
	 }
	 
	 public static int size(int k){
		 return BDMData.get(k).getSize();
	 }
	 
	 public static int size(int k, int i){
		 //return BDMData.get(k).getSize();
		 Object partitionSize = BDMData.get(k).getPartitions().get(i);
		 if (partitionSize != null) return (Integer) partitionSize;
		 return 0;
	 }
	 
	 public static String blockKey(int blockIndex){
		 return BDMData.get(blockIndex).getKey();
	 }
	 
	 public static int blockIndex(String blockingkey){
		 return blockingkeyIndexer.get(blockingkey);
	 }
	 
	public static int getTotalToPatition(String blockingkey, int partition){
		HashMap<Integer, Integer> partitions = BDMData.get(blockIndex(blockingkey)).getPartitions();
		Object[] partitionsSet = partitions.keySet().toArray();
		int entitiesCount = 0;
		for (Object object : partitionsSet) {
			if ((Integer)object < partition){
				entitiesCount += partitions.get(object); 
			}
		}
		return entitiesCount;
	}
	 
}